{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Spark Basics 1\n",
    "\n",
    "This notebook introduces two fundamental objects in Spark:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "* The Spark Context"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "* The Resilient Distributed DataSet or RDD"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Spark Context\n",
    "We start by creating a **SparkContext** object named **sc**. In this case we create a spark context that uses 4 *executors* (one per core)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<SparkContext master=local[4] appName=pyspark-shell>\n"
     ]
    }
   ],
   "source": [
    "#start the SparkContext\n",
    "import findspark\n",
    "findspark.init()\n",
    "\n",
    "from pyspark import SparkContext \n",
    "sc = SparkContext(master=\"local[4]\")\n",
    "print(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Only one sparkContext at a time!\n",
    "* Spark is designed for single user\n",
    "* Only one sparkContext per program/notebook.\n",
    "* Before starting a new sparkContext. Stop the one currently running"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true,
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "# sc.stop() #commented out so that you don't stop your context by mistake"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## RDDs\n",
    "\n",
    "<p>RDD (or Resilient Distributed DataSet) is the main novel data structure in Spark. You can think of it as a list whose elements are stored on several computers.</p>\n",
    "\n",
    "<p><img alt=\"\" src=\"Figures/SparkContextAndRDD.jpg\" style=\"height:324px; width:900px\" /></p>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "The elements of each `RDD` are distributed across the **worker nodes** which are the nodes that perform the actual computations. This notebook, however, is running on the **Driver node**. As the RDD is not stored on the driver-node you cannot access it directly. The variable name `RDD` is really just a pointer to a python object which holds the information regardnig the actual location of the elements."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Some basic RDD commands"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Parallelize \n",
    "* Simplest way to create an RDD.\n",
    "* The method `A=sc.parallelize(L)`, creates an RDD named `A` from list `L`.\n",
    "* `A` is an RDD of type `PythonRDD`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[1] at RDD at PythonRDD.scala:48"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "A=sc.parallelize(range(3))\n",
    "A"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Collect\n",
    "\n",
    "* RDD content is distributed among all executors.\n",
    "* `collect()` is the inverse of `parallelize()'\n",
    "* collects the elements of the RDD\n",
    "* Returns a list\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'list'>\n",
      "[0, 1, 2]\n"
     ]
    }
   ],
   "source": [
    "L=A.collec t()\n",
    "print(type(L))\n",
    "print(L)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "#### Using `.collect()` eliminates the benefits of parallelism\n",
    "It is often tempting to `.collect()` and RDD, make it into a list, and then process the list using standard python. However, note that this means that you are using only the head node to perform the computation which means that you are not getting any benefit from spark.\n",
    "\n",
    "Using RDD operations, as described below, **will** make use of all of the computers at your disposal."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Map\n",
    "* applies a given operation to each element of an RDD\n",
    "* parameter is the function defining the operation.\n",
    "* returns a new RDD.\n",
    "* Operation performed in parallel on all executors.\n",
    "* Each executor operates on the data **local** to it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[0, 1, 4]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "A.map(lambda x: x*x).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "**Note:** Here we are using **lambda** functions, later we will see that regular functions can also be used.\n",
    "\n",
    "For more on lambda function see [here](http://www.secnetix.de/olli/Python/lambda_functions.hawk)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Reduce\n",
    "\n",
    "* Takes RDD as input, returns a single value.\n",
    "* **Reduce operator** takes **two** elements as input returns **one** as output.\n",
    "* Repeatedly applies a **reduce operator**\n",
    "* Each executor reduces the data local to it.\n",
    "* The results from all executors are combined."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "The simplest example of a 2-to-1 operation is the sum:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "A.reduce(lambda x,y:x+y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Here is an example of a reduce operation that finds the shortest string in an RDD of strings."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'is'"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "words=['this','is','the','best','mac','ever']\n",
    "wordRDD=sc.parallelize(words)\n",
    "wordRDD.reduce(lambda w,v: w if len(w)<len(v) else v)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "## Properties of reduce operations\n",
    "\n",
    "* Reduce operations **must not depend on the order**\n",
    "  * Order of operands should not matter\n",
    "  * Order of application of reduce operator should not matter\n",
    "\n",
    "* Multiplication and summation are good:\n",
    "\n",
    "```\n",
    "                1 + 3 + 5 + 2                      5 + 3 + 1 + 2 \n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    " * Division and subtraction are bad:\n",
    "\n",
    "```\n",
    "                    1 - 3 - 5 - 2                      1 - 3 - 5 - 2\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "### Why must reordering not change the result?\n",
    "\n",
    "You can think about the reduce operation as a binary tree where the leaves are the elements of the list and the root is the final result. Each triplet of the form (parent, child1, child2) corresponds to a single application of the reduce function. \n",
    "\n",
    "The order in which the reduce operation is applied is **determined at run time** and depends on how the RDD is partitioned across the cluster.\n",
    "There are many different orders to apply the reduce operation. \n",
    "\n",
    "If we want the input RDD to uniquely determine the reduced value **all evaluation orders must must yield the same final result**. In addition, the order of the elements in the list must not change the result. In particular, reversing the order of the operands in a reduce function must not change the outcome. \n",
    "\n",
    "For example the arithmetic operations multiply `*` and add `+` can be used in a reduce, but the operations subtract `-` and divide `/` should not.\n",
    "\n",
    "Doing so will not raise an error, but the result is unpredictable."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "-9"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "B=sc.parallelize([1,3,5,2])\n",
    "B.reduce(lambda x,y: x-y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "Which of these the following orders was executed?\n",
    "* $$((1-3)-5)-2$$ or\n",
    "* $$(1-3)-(5-2)$$"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Using regular functions instead of lambda functions\n",
    "\n",
    "* lambda function are short and sweet.\n",
    "* but sometimes it's hard to use just one line.\n",
    "* We can use full-fledged functions instead."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "A.reduce(lambda x,y: x+y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Suppose we want to find the \n",
    "* last word in a lexicographical order \n",
    "* among \n",
    "* the longest words in the list.\n",
    "\n",
    "We could achieve that as follows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "run_control": {
     "frozen": false,
     "read_only": false
    },
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'this'"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def largerThan(x,y):\n",
    "    if len(x)>len(y): return x\n",
    "    elif len(y)>len(x): return y\n",
    "    else:  #lengths are equal, compare lexicographically\n",
    "        if x>y: \n",
    "            return x\n",
    "        else: \n",
    "            return y\n",
    "        \n",
    "wordRDD.reduce(largerThan)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Summary\n",
    "\n",
    "We saw how to:\n",
    "* Start a SparkContext\n",
    "* Create an RDD\n",
    "* Perform Map and Reduce operations on an RDD\n",
    "* Collect the final results back to head node."
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "celltoolbar": "Slideshow",
  "hide_input": false,
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.3"
  },
  "toc": {
   "colors": {
    "hover_highlight": "#DAA520",
    "running_highlight": "#FF0000",
    "selected_highlight": "#FFD700"
   },
   "moveMenuLeft": true,
   "nav_menu": {
    "height": "512px",
    "width": "252px"
   },
   "navigate_menu": true,
   "number_sections": true,
   "sideBar": true,
   "threshold": 4,
   "toc_cell": false,
   "toc_section_display": "block",
   "toc_window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
